package com.com.zs21cp.ApiDemo;

import com.com.zs21cp.ToolBean.WaterSensor;
import com.com.zs21cp.ToolBean.WaterSensorMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class ProcessDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = env.socketTextStream("hadoop2", 10086);
        OutputTag<WaterSensor> s1 = new OutputTag<>("s1", Types.POJO(WaterSensor.class));
        OutputTag<WaterSensor> s2 = new OutputTag<>("s2", Types.POJO(WaterSensor.class));
        //process后的ProcessFunction带的参数第一个是In，第二个是OUT，第一个不用动，保持原样就行，第二个需要修改成
        //你想要输出的数据流的格式。
        SingleOutputStreamOperator<String> process = source
            .map(new WaterSensorMapFunction())
            .process(new ProcessFunction<WaterSensor, String>() {
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                    if (value.id.equals("s1")) {
                        ctx.output(s1, value);
                    } else if (value.id.equals("s2")) {
                        ctx.output(s2, value);
                    } else {
                        out.collect(value.id);
                    }
                }
            });
        process.print("主流：");
        SideOutputDataStream<WaterSensor> sideOutput = process.getSideOutput(s1);
        SideOutputDataStream<WaterSensor> sideOutput2 = process.getSideOutput(s2);
        sideOutput.print("测输出流1：");
        sideOutput2.print("测输出流2：");


        env.execute();
    }
}
